package com.iqik.rocket.config;

import com.iqik.rocket.dao.OrderRepository;
import com.iqik.rocket.pojo.Order;
import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.client.producer.SendCallback;
import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.common.message.Message;
import org.apache.rocketmq.common.message.MessageQueue;
import org.apache.rocketmq.spring.annotation.RocketMQMessageListener;
import org.apache.rocketmq.spring.core.RocketMQListener;
import org.apache.rocketmq.spring.core.RocketMQTemplate;
import org.apache.rocketmq.spring.support.RocketMQUtil;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

@Slf4j
@Component
@RocketMQMessageListener(topic = "tp_order_02", consumerGroup = "consume_order")
public class OrderRocketListener implements RocketMQListener<Order> {

    @Autowired
    private RocketMQTemplate rocketMQTemplate;

    @Autowired
    private OrderRepository orderRepository;

    @Override
    public void onMessage(Order order) {
        log.info("收到并存储：" + order.toString());
        orderRepository.save(order);
        String id = order.getId();
        Message message = new Message("tp_pay_02", id.getBytes());
        org.springframework.messaging.Message<String> newMessage = RocketMQUtil.convertToSpringMessage(message);
        rocketMQTemplate.asyncSend(message.getTopic(), newMessage, new SendCallback() {

            @Override
            public void onSuccess(SendResult sendResult) {
                log.info("发送订单id到待支付队列" + sendResult.getMsgId());
            }

            @Override
            public void onException(Throwable throwable) {
            }
        }, 3000, 5);
    }
}
